package net.wlm.others;


import net.wlm.utils.DateTimeUtil;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class BatteryHealthWindowFunction implements WindowFunction<BatteryHealthData, String, String, TimeWindow> {
    @Override
    public void apply(String s, TimeWindow window, Iterable<BatteryHealthData> input, Collector<String> out) throws Exception {
        String startTime = DateTimeUtil.convertLongToString(window.getStart(), DateTimeUtil.DATE_TIME_FORMAT);
        String endTime = DateTimeUtil.convertLongToString(window.getEnd(), DateTimeUtil.DATE_TIME_FORMAT);
        double count = 0;
        double sum = 0;
        for (BatteryHealthData element : input){
            count++;
            sum+=element.getFullChargeCapacity();
        }
        String output = startTime + "," + endTime + ","
                + count + "," + sum + "," + (sum/count) + ","
                + System.currentTimeMillis();
        out.collect(output);

    }
}